package trunk.parallel.sameas;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Set;

import com.hp.hpl.jena.graph.Node;
import com.hp.hpl.jena.graph.Triple;
import com.hp.hpl.jena.sparql.core.Var;
import com.hp.hpl.jena.sparql.engine.binding.Binding;
import com.hp.hpl.jena.sparql.engine.binding.BindingFactory;
import trunk.config.Config;
import trunk.description.Statistics;
import trunk.description.RemoteService;
import trunk.graph.Edge;
import trunk.graph.SimpleEdge;
import trunk.graph.Vertex;
import trunk.parallel.engine.exec.TripleExecution;
import trunk.parallel.engine.main.StageGen;
import trunk.stream.engine.util.HypTriple;

public class SameAsStageGen extends StageGen {

	public static final String sameAs = "http://www.w3.org/2002/07/owl#sameAs";
	public SameAsStageGen(Statistics config) {
		super(config);
	}

	@Override
	protected List<SimpleEdge> make(Collection<Triple> triples) {
		
		List<SimpleEdge> edges = super.make(triples);
		for(Edge e:edges) {
			transform((SimpleEdge)e);
		}
		return edges;
	}
	
	
	/**
	 * Identify the nodes having equivalent URIs and replace them with variables.
	 * Also combine all sameAs URIs to generate initial bindings.
	 * @param t
	 * @return
	 */
	private void transform(SimpleEdge e) {
		Vertex v1 = e.getV1();
		Vertex v2 = e.getV2();
		
		if(findSA(v1) || findSA(v2)) {
			HypTriple t = new HypTriple(new Triple(v1.getNode(),e.getTriple().getPredicate(),v2.getNode()),e.getTriple().getServices());
			e.setTriple(t);
		}
	}
	
	private boolean findSA(Vertex v) {
		Node n = (Node) v.getNode();
		if(n.isConcrete()) {
			Set<Binding> bindings = sameAsQuery(n);
			if(bindings.size() == 1)
				return false;
			if(Config.debug) {
				System.out.println("SA of "+n.getLocalName()+": "+(bindings.size()-1));
			}
			Node node = Var.alloc(n.getLocalName()+"sa");
			v.setNode(node);
			v.setBindings(bindings);
			return true;
		}
		return false;
	}

	/**
	 * Get equivalent URIs for a given node.
	 * @param n
	 * @return 
	 */
	private Set<Binding> sameAsQuery(Node n) {
		Triple tri = new Triple(n, Node.createURI(sameAs), Var.alloc(n.getLocalName()+"sa"));
		//List<RemoteService> services = config.getServiceRegistry().getMatchingServices(tri);
		List<RemoteService> services = new ArrayList<RemoteService>();
		services.addAll(config.getServiceRegistry().getAvailableServices());
		services.add(new RemoteService("http://152.78.189.45:8080/openrdf-sesame/repositories/sameas"));
		services.add(new RemoteService("http://152.78.189.148:8080/openrdf-sesame/repositories/sameas"));
		TripleExecution te = new TripleExecution(new HypTriple(tri,services));
		Set<Binding> bindings = te.exec(null);
		//add the initial value of this node
		Binding originalValue = BindingFactory.create();
		originalValue.add(Var.alloc(n.getLocalName()+"sa"), n);
		bindings.add(originalValue);
		return bindings;
	}
	
}
